package com.aaa.shuke2356.consumer;

import com.aaa.shuke2356.entity.Status;
import com.aaa.shuke2356.service.StatusService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author ：Teacher陈（86521760@qq.com）
 * @date ：Created in 2024/11/25 10:31
 * @description：我的消费类
 * @modified By：
 * @version: 1.0
 */
@Component
public class MyConsumer {
    @Autowired
    private StatusService statusService;
    //KafkaListener是消息队列的监听器
    @KafkaListener(topics = "chenjian")
    public void consumeMessage(ConsumerRecord record){
        //打印数据
        System.out.println(new Date()+"消费的数据："+record.value());
        //将json字符串转换成java对象
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            Status status = objectMapper.readValue(record.value().toString(), Status.class);
            //将数据进行持久化，通过mybatis持久层框架进行
            statusService.insert(status);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }


    }
}
